/*
 * =====================================================================================
 *
 *       Filename:  cl_event_looper.cxx
 *
 *    Description:  
 *
 *        Version:  1.0
 *        Created:  04/07/2011 09:17:04 AM
 *       Revision:  none
 *       Compiler:  gcc
 *
 *         Author:  Ken Chen (Ken), chen_kenz@emc.com
 *        Company:  EMC
 *
 * =====================================================================================
 */

#ifndef __GXX_EXPERIMENTAL_CXX0X__
#define __GXX_EXPERIMENTAL_CXX0X__ 
#endif

#include <sys/eventfd.h>
#include <cstdint>
#include <cassert>
#include "cl_event_looper.hxx"
#include "cl_id.hxx"
#include "cl_mutex.hxx"
#include "cl_lock_guard.hxx"
#include "cl_channel.hxx"
#include "cl_thread.hxx"
#include "cl_poller.hxx"
#include "cl_poller_factory.hxx"
#include "cl_logger.hxx"
#include "cl_timestamp.hxx"
#include "cl_timer_service.hxx"

namespace neo
{
using namespace std; 

namespace
{
const int pollInterval = 10000;
}

CLEventLooper::CLEventLooper()
  : m_eventfd(::eventfd(0, EFD_CLOEXEC | EFD_NONBLOCK)),
  m_shutdown(false),
  m_lock(new CLMutex),
  m_poller(CLPollerFactory::create()),
  m_eventChannel(new CLChannel(m_eventfd, CLChannel::CL_READ_EVENT)),
	m_timerService(new CLTimerService)
{
  m_thr.reset(new CLThread(bind(&CLEventLooper::loop, this)));
  m_eventChannel->setReadCallback(bind(&CLEventLooper::handleRead, this));
}

CLEventLooper::~CLEventLooper()
{
  shutdown();
  if(m_eventfd >= 0)
  {
    ::close(m_eventfd);
  }
}

void CLEventLooper::start(void)
{
  m_thr->start();
}

void CLEventLooper::loop(void)  
{
  addChannel(m_timerService->timerChannel());
  addChannel(m_eventChannel.get());
  CLChannelList::iterator itr, itrEnd;
  CLChannelList readyChannels;
  while(!m_shutdown)    
  {
    readyChannels.clear(); 
    m_poller->poll(nextPollInterval(), readyChannels);
    itr = readyChannels.begin();
    itrEnd = readyChannels.end();
    for(; itr != itrEnd; ++itr)
    {
      assert(*itr != NULL);
      (*itr)->handleEvent();
    }
    doPendingJobs();
  }
}

void CLEventLooper::shutdown(void)
{
	if(m_shutdown)
	{
		return;
	}
  m_shutdown = true;
  wakeup();
  m_thr->join();
}

void CLEventLooper::handleRead(void)
{
  uint64_t u = 0; 
  if(::read(m_eventfd, &u, sizeof(u)) != sizeof(u))
  {
	  CLLOG_E("POLLOER", "CLEventLooper::handleRead %s", 
        "failed to read wakeup event");
  }
}

void CLEventLooper::runJob(const CLCallbackFunctor& cb)
{
  if(isInLooperThread()) 
	{
    cb();
	}
  else
  {
	  queueJob(cb);
    wakeup();
  }
}

void CLEventLooper::queueJob(const CLCallbackFunctor& cb)
{
 	{
    CLLockGuard guard(*m_lock);
    m_callbacks.push_back(cb);
  }

	wakeup();
}

// interact with Timer service
const CLId CLEventLooper::runJobAt(const CLTimestamp& atTime, 
                                   const CLCallbackFunctor& timerCallback) 
{
  return m_timerService->schedule(timerCallback, atTime, 0);
}

const CLId CLEventLooper::runJobAfter(int msecsDelay, 
                                      const CLCallbackFunctor& timerCallback) 
{
	return m_timerService->schedule(timerCallback, 
                                  CLTimestamp::nowTime() + msecsDelay, 
                                  0);
}

const CLId CLEventLooper::runJobEvery(int msecsInterval, 
                                      const CLCallbackFunctor& timerCallback) 
{
  return m_timerService->schedule(timerCallback, 
                                  CLTimestamp::nowTime(), 
                                  msecsInterval);
}

void CLEventLooper::cancelTimerJob(const CLId& timerId) 
{
  return  m_timerService->removeTimer(timerId);
}

bool CLEventLooper::addChannel(CLChannel* ch) 
{
//  CLLockGuard guard(*m_lock);
  assert(isInLooperThread());
  return m_poller->add(ch);
}

bool CLEventLooper::removeChannel(CLChannel* ch) 
{
// CLLockGuard guard(*m_lock);
  assert(isInLooperThread());
  return m_poller->remove(ch);
}

bool CLEventLooper::modifyChannel(CLChannel* ch) 
{
//  CLLockGuard guard(*m_lock);
  assert(isInLooperThread());
  return m_poller->modify(ch);
}

void CLEventLooper::wakeup(void)
{
  uint64_t u = 1;
  if(::write(m_eventfd, &u, sizeof(u)) != sizeof(u))
  {
    CLLOG_E("POLLER", "CLEventLooper::wakeup %s", 
        "failed to wakeup eventlooper");
  }
}

void CLEventLooper::doPendingJobs(void)
{
  CallbackVector callbacks;
  {
    CLLockGuard guard(*m_lock);
    callbacks.swap(m_callbacks);
  }

  CallbackVector::iterator itr = callbacks.begin();
  CallbackVector::iterator itrEnd = callbacks.end();
  for(; itr != itrEnd; ++itr)
  {
    (*itr)();
  }
}

inline int CLEventLooper::nextPollInterval(void) const 
{
	const CLTimestamp nextExpiredTime(m_timerService->nextExpiredTime());
  if(nextExpiredTime == CLTimestamp::nullTime())
	{
		CLLOG_D("POLLER", "Got default poll interval %d", pollInterval);
    return pollInterval;
	}
	else
	{
    int secs = 0, msecs = 0;
		clTimestampDiff(nextExpiredTime, 
				            CLTimestamp::nowTime(),
										secs,
										msecs);
		CLLOG_D("POLLER", "Got next poll interval from timer service %d", 
        ((secs*1000000) + msecs));
    return (secs*1000000) + msecs;
	}
}

} // neo

